﻿#include "mongodb_bulk_task.h"
#include "mongodb_mgr.h"
#include "detail/mongodb_bulk_data.h"
#include "detail/mongodb_client.h"
#include "log.h"

MongoDbBulkTask::MongoDbBulkTask(const std::string &instName, const std::string &dbName, const std::string &collectionName, unsigned type, const std::string &queryString, const std::string &docString)
	: m_instName(instName)
	, m_dbName(dbName)
	, m_collectionName(collectionName)
	, m_type(type)
	, m_queryString(queryString)
	, m_docString(docString)
	, m_bFirst(true)
{
	m_taskInstName = "MongoDbBulkTask_" + m_instName;
	m_id = MongoDbBulkData::GetId(m_instName, m_dbName, m_collectionName);
}

MongoDbBulkTask::~MongoDbBulkTask()
{

}

void MongoDbBulkTask::Process(void *sharedObj)
{
	if (sharedObj == nullptr)
	{
		LOG_ERROR_TO("MongoDbBulkTask", Fmt("task process failure, no find mongodb obj"));
		return;
	}
	MongoDbClient *client = (MongoDbClient *)sharedObj;
	bool createRet = client->BulkCreate(m_dbName, m_collectionName);
	if (!createRet)
	{
		return;
	}

	if (m_datas)
	{
		auto &datas = *m_datas.get();
		for (size_t i = 0; i < datas.size(); i++)
		{
			switch (datas[i].type)
			{
			case e_bulkopt_remove:
				client->BulkOptRemove(datas[i].query);
				break;
			case e_bulkopt_insert:
				client->BulkOptInsert(datas[i].doc);
				break;
			case e_bulkopt_update:
				client->BulkOptUpdate(datas[i].query, datas[i].doc, true);
				break;
			case e_bulkopt_replace:
				client->BuikOptReplace(datas[i].query, datas[i].doc, true);
				break;
			}
		}
	}
	client->BulkExecute();
}

void MongoDbBulkTask::OnCompleted()
{
	MongoDbBulkData::get_mutable_instance().SetDataState(m_id, e_bulkopt_state_idle);
	auto task = std::make_shared<MongoDbBulkTask>(m_instName, m_dbName, m_collectionName, 0, "", "");
	task->m_bFirst = false;
	if (m_cb)	//for test
	{
		task->m_cb = m_cb;
		LOG_INFO_TO("mongodb test:", " insert some record completed ...");
	}
	task->Go();
}

void *MongoDbBulkTask::CreateSharedObj(unsigned index)
{
	auto &objs = MongoDbMgr::get_mutable_instance().GetDbInstance(m_instName + "_bulk");
	if (index >= objs.size())
	{
		return nullptr;
	}
	return objs[index].get();
}

void MongoDbBulkTask::Go()
{
	if (m_bFirst)
	{
		m_bFirst = false;
		switch (m_type)
		{
		case e_bulkopt_remove:
			MongoDbBulkData::get_mutable_instance().BulkOptRemove(m_id, m_queryString);
			break;
		case e_bulkopt_insert:
			MongoDbBulkData::get_mutable_instance().BulkOptInsert(m_id, m_docString);
			break;
		case e_bulkopt_update:
			MongoDbBulkData::get_mutable_instance().BulkOptUpdate(m_id, m_queryString, m_docString);
			break;
		case e_bulkopt_replace:
			MongoDbBulkData::get_mutable_instance().BuikOptReplace(m_id, m_queryString, m_docString);
			break;
		}
	}

	unsigned state = MongoDbBulkData::get_mutable_instance().GetDataState(m_id);
	if (state == e_bulkopt_state_idle)
	{
		auto &datas = MongoDbBulkData::get_mutable_instance().GetData(m_id);
		if (!datas.empty())
		{
			m_datas = std::make_shared<DataSetType>(std::move(datas));
			MongoDbBulkData::get_mutable_instance().RemoveData(m_id);
			MongoDbBulkData::get_mutable_instance().SetDataState(m_id, e_bulkopt_state_working);
			MultiTask::TaskBase::Go();
		}
		else
		{
			// for test
			if (m_cb)
			{
				m_cb();
			}
		}
	}
}


MongoDbBulkTask_Remove::MongoDbBulkTask_Remove(const std::string &instName, const std::string &dbName, const std::string &collectionName, const std::string &queryString)
	: MongoDbBulkTask(instName, dbName, collectionName, e_bulkopt_remove, queryString, "")
{}

MongoDbBulkTask_Insert::MongoDbBulkTask_Insert(const std::string &instName, const std::string &dbName, const std::string &collectionName, const std::string &docString)
	: MongoDbBulkTask(instName, dbName, collectionName, e_bulkopt_insert, "", docString)
{}


MongoDbBulkTask_Update::MongoDbBulkTask_Update(const std::string &instName, const std::string &dbName, const std::string &collectionName, const std::string &queryString, const std::string &docString)
	: MongoDbBulkTask(instName, dbName, collectionName, e_bulkopt_update, queryString, docString)
{}
